package com.zr.influx.service;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import com.zr.influx.model.ForexDataDTO;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

@Service
public class ForexQueryServiceImpl implements ForexQueryService {

    private final InfluxDBClient influxDBClient;
    private final String bucket;
    private final String org;

    public ForexQueryServiceImpl(InfluxDBClient influxDBClient,
                                 @Value("${influxdb.bucket}") String bucket,
                                 @Value("${influxdb.org}") String org) {
        this.influxDBClient = influxDBClient;
        this.bucket = bucket;
        this.org = org;
    }

    @Override
    public List<String> getAllCurrencyPairs() {
        String flux = String.format("from(bucket: \"%s\") " +
                "|> range(start: -30d) " +
                "|> filter(fn: (r) => r._measurement == \"forex\") " +
                "|> keep(columns: [\"currencyPair\"]) " +
                "|> distinct(column: \"currencyPair\")", bucket);

        List<FluxTable> tables = influxDBClient.getQueryApi().query(flux, org);

        return tables.stream()
                .flatMap(table -> table.getRecords().stream())
                .map(record -> record.getValueByKey("currencyPair").toString())
                .distinct()
                .collect(Collectors.toList());
    }

    @Override
    public Optional<ForexDataDTO> getLatestPrice(String currencyPair) {
        String flux = "from(bucket: \"tick\") \n" +
                "|> range(start: -1h) \n" +
                "|> filter(fn: (r) => r._measurement == \"forex\" and r.currencyPair == \"EUR/USD\") \n" +
                "|> last() \n" +
                "|> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")";

        List<FluxTable> tables = influxDBClient.getQueryApi().query(flux, org);
        if (tables.isEmpty() || tables.get(0).getRecords().isEmpty()) {
            return Optional.empty();
        }
        FluxRecord record = tables.get(0).getRecords().get(0);
        return Optional.of(convertToDTO(record));
    }

    @Override
    public List<ForexDataDTO> getPricesInRange(String currencyPair, Instant startTime,
                                               Instant endTime, String exchange) {
        StringBuilder fluxBuilder = new StringBuilder(String.format(
                "from(bucket: \"%s\") " +
                        "|> range(start: %d, stop: %d) " +
                        "|> filter(fn: (r) => r._measurement == \"forex\" and " +
                        "r.currencyPair == \"%s\") ",
                bucket, startTime.toEpochMilli() * 1_000_000,
                endTime.toEpochMilli() * 1_000_000, currencyPair));

        // 如果指定了交易所，添加交易所过滤条件
        if (exchange != null && !exchange.isEmpty()) {
            fluxBuilder.append(String.format("and r.exchange == \"%s\" ", exchange));
        }

        fluxBuilder.append("|> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\") ");
        fluxBuilder.append("|> sort(columns: [\"_time\"])");

        List<FluxTable> tables = influxDBClient.getQueryApi().query(fluxBuilder.toString(), org);

        return convertRecordsToDTOs(tables);
    }

    @Override
    public List<ForexDataDTO> getAggregatedData(String currencyPair, Instant startTime,
                                                Instant endTime, String interval) {
        String flux = String.format("from(bucket: \"%s\") " +
                        "|> range(start: %d, stop: %d) " +
                        "|> filter(fn: (r) => r._measurement == \"forex\" and " +
                        "r.currencyPair == \"%s\" and " +
                        "r._field =~ /(openPrice|highPrice|lowPrice|closePrice|volume)/) " +
                        "|> window(every: %s) " +
                        "|> aggregateWindow(every: %s, fn: mean, createEmpty: false) " +
                        "|> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\") " +
                        "|> sort(columns: [\"_time\"]) ",
                bucket, startTime.toEpochMilli() * 1_000_000,
                endTime.toEpochMilli() * 1_000_000, currencyPair, interval, interval);
        List<FluxTable> tables = influxDBClient.getQueryApi().query(flux, org);
        return convertRecordsToDTOs(tables);
    }

    /**
     * 将FluxRecord转换为ForexDataDTO
     */
    private ForexDataDTO convertToDTO(FluxRecord record) {
        ForexDataDTO dto = new ForexDataDTO();
        dto.setCurrencyPair(record.getValueByKey("currencyPair").toString());
        dto.setExchange(record.getValueByKey("exchange").toString());
        dto.setTimestamp(record.getTime());

        // 设置价格字段
        if (record.getValueByKey("openPrice") != null) {
            dto.setOpenPrice(((Number) record.getValueByKey("openPrice")).doubleValue());
        }
        if (record.getValueByKey("highPrice") != null) {
            dto.setHighPrice(((Number) record.getValueByKey("openPrice")).doubleValue());
        }
        if (record.getValueByKey("lowPrice") != null) {
            dto.setLowPrice(((Number) record.getValueByKey("openPrice")).doubleValue());
        }
        if (record.getValueByKey("closePrice") != null) {
            dto.setClosePrice(((Number) record.getValueByKey("openPrice")).doubleValue());
        }
        if (record.getValueByKey("volume") != null) {
            dto.setVolume(((Number) record.getValueByKey("volume")).longValue());
        }

        return dto;
    }

    /**
     * 将查询结果转换为DTO列表
     */
    private List<ForexDataDTO> convertRecordsToDTOs(List<FluxTable> tables) {
        List<ForexDataDTO> result = new ArrayList<>();

        for (FluxTable table : tables) {
            for (FluxRecord record : table.getRecords()) {
                result.add(convertToDTO(record));
            }
        }

        return result;
    }
}

